package org.study.canal.api;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * 参考 https://www.bbsmax.com/A/B0zq0ZYndv
 *
 * @author Administrator
 * @date 2021-08-16
 */
public class CanalTest02 {
    private static Logger log = LoggerFactory.getLogger(CanalTest02.class);

    private static final int BATCH_SIZE = 1000;
    private static final String HOSTNAME = "127.0.0.1";
    private static final int PORT = 11111;
    private static final String DESTINATION = "example";
    private static final String USERNAME = "";
    private static final String PASSWORD = "";

    public static void main(String[] args) {
        new Thread(() -> initConnector("example")).start();
        new Thread(() -> initConnector("example")).start();
    }

    private static void initConnector(String destination) {
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(HOSTNAME, PORT), destination, USERNAME, PASSWORD);
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            while (true) {
                Message message = connector.getWithoutAck(1000);
                if (message.getId() != -1 && message.getEntries().size() > 0) {
                    printEntry(message.getEntries());
                }
                connector.ack(message.getId());
            }
        } finally {
            connector.disconnect();
        }
    }


    /**
     * 打印 Canal Server 解析 binlog 获得的实体类信息
     *
     * @param entryList
     */
    private static void printEntry(List<CanalEntry.Entry> entryList) {
        for (CanalEntry.Entry entry : entryList) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                // 开启、关闭事务的实体类型：跳过
                continue;
            }
            CanalEntry.RowChange rowChange = null;
            try {
                // 获取发生变化的数据：RowChange 对象包含了一行数据变化的所有特征
                // 如 isDdl 表示是否为 ddl 操作，beforeColumns 表示变更前的数据字段，afterColumns 表示变更后的数据字段等
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("解析数据异常，当前数据：" + entry.toString(), e);
            }
            // 获取事件类型：INSERT、UPDATE、DELETE 类型
            CanalEntry.EventType eventType = rowChange.getEventType();
            log.debug("判断是否为 DDL 语句。isDdl={}，sql={}", rowChange.getIsDdl(), rowChange.getSql());

            CanalEntry.Header header = entry.getHeader();
            String formatMsg = String.format("logfileName=%s，logfileOffset=%s，schemaName=%s，tableName=%s，eventType=%s",
                    // 获取发生变化的 binlog 名称
                    header.getLogfileName(),
                    // 获取发生变化的 binlog 偏移量
                    header.getLogfileOffset(),
                    // 获取发生变化的数据库名称
                    header.getSchemaName(),
                    // 获取发生变化的表名称
                    header.getTableName(),
                    // 获取事件类型
                    eventType);
            log.info(formatMsg);

            // 获取 RowChange 对象里的每一行数据
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                // 获取变更前的数据
                List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                // 获取变更后的数据
                List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                if (eventType == CanalEntry.EventType.INSERT) {
                    // 如果是 INSERT 语句
                    printColumn(afterColumnsList);
                } else if (eventType == CanalEntry.EventType.DELETE) {
                    // 如果是 DELETE 语句
                    printColumn(beforeColumnsList);
                } else if (eventType == CanalEntry.EventType.UPDATE) {
                    // 如果是 UPDATE 语句
                    System.out.println("【before update】");
                    printColumn(beforeColumnsList);
                    System.out.println("【after update】");
                    printColumn(afterColumnsList);
                } else {
                    // 其他语句
                    log.warn("其他(除 INSERT、DELETE、UPDATE 外)类型的事件");
                }
            }
        }
    }

    private static void printColumn(List<CanalEntry.Column> columnList) {
        for (CanalEntry.Column column : columnList) {
            System.out.println(column.getName() + "：" + column.getValue() + "，update=" + column.getUpdated());
        }
    }
}
